package com.chinaoryx.er;

import org.apache.log4j.Logger;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

/**
 * Created by IntelliJ IDEA.
 * User: hyman.zhang
 * Date: 12-2-4
 * Time:
 */
public class TransferThread implements Runnable {

    static Logger logger = Logger.getLogger(TransferThread.class);
    
    private int batchCount = 40;

    private String threadId = null;
    private Connection conn1;
    private Connection conn2;
    EntityDefination entity;
    TransferThreadManager manager = null;
    BlockingQueue<Map<String, Object>> queue = null;


    public TransferThread(String threadId, Connection conn1, Connection conn2, EntityDefination entity, TransferThreadManager manager, int batchCount) throws SQLException {
        this.threadId = threadId;
        this.conn1 = conn1;
        this.conn2 = conn2;
        this.entity = entity;
        this.manager = manager;
        this.batchCount = batchCount;
        this.queue = this.manager.getQueue();
        Statement stmt = conn2.createStatement();
        stmt.execute("set foreign_key_checks=0");
        stmt.close();
    }

    @Override
    public void run() {
        logger.info("thread(" + this.threadId + ") started.");
        DataTransfer dataTransfer = new DataTransfer();
        Map<String, Object> record = null;





        try {
            record = this.queue.take();
        } catch (InterruptedException e) {
            logger.error("LOG00010:", e);

        }

        int count = 0;
        Map<String, PreparedStatement> pstmtMap = new LinkedHashMap<String, PreparedStatement>();

        while (record != null) {
            try {

                entity.getTable().moveTableRecord(conn1, conn2, record, pstmtMap);
                logger.info("thread(" + this.threadId + ") process ("+count+").");
                count++;
            } catch (SQLException e) {
                logger.error("LOG00020:", e);

            }

            // batch execute
            if (count == batchCount) {
                count = 0;
                for (Map.Entry<String, PreparedStatement> entry : pstmtMap.entrySet()) {
                    try {
                        entry.getValue().executeBatch();
                        conn2.commit();
//                        entry.getValue().close();
                        logger.info("thread(" + this.threadId + ") batch executed .");
                    } catch (SQLException e) {
                        logger.error("LOG00050:", e);
                        logger.info("thread(" + this.threadId + ") done .");
                        return;
                    }

                }
            }
            // quit...
            if (this.manager.isDone() && this.queue.size() == 0) {
                for (Map.Entry<String, PreparedStatement> entry : pstmtMap.entrySet()) {
                    try {
                        entry.getValue().executeBatch();
                        entry.getValue().close();
                        conn2.commit();
                    } catch (SQLException e) {
                        logger.error("LOG00050:", e);
                        logger.info("It's done!");
                        return;
                    }
                    logger.info("thread(" + this.threadId + ") done .");
                    return;
                }
            }

            // get next
            try {
                record = this.queue.take();
            } catch (InterruptedException e) {
                logger.error("LOG00040:", e);
            }

        }


    }
}
